package com.fengye.register;

import com.alibaba.fastjson.JSON;
import com.fengye.common.IPHelper;
import com.fengye.config.PropertyConfigHelper;
import com.fengye.model.InvokeService;
import com.fengye.model.ProviderService;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.ClassUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

/**
 * 注册中心，包含服务治理、服务提供者注册、服务消费者注册
 *
 * 后续需要拆分开来：服务中心当作容器，存放服务消费和服务提供者缓存，对于服务提供者和消费者提供单独的接口
 *
 * @Author fengye
 * @Date 2023/2/13 19:25
 **/
public class RegisterCenter implements RegisterCenter4Governance, RegisterCenter4Provider,RegisterCenter4Invoker{

    private static Logger logger = LoggerFactory.getLogger(RegisterCenter.class);

    /** 单例注册中心 */
    private static RegisterCenter registerCenter = new RegisterCenter();

    /** 服务提供者列表，key:服务提供者接口， value: 服务提供者服务方法列表 */
    private final static Map<String, List<ProviderService>> providerServiceMap = Maps.newConcurrentMap();

    /** 服务端ZK服务信息，选择服务（第一次直接从ZK拉起，之后由ZK的监听机制主动更新） */
    private final static Map<String, List<ProviderService>> serviceMetaData4ConsumerMap = Maps.newConcurrentMap();

    private static String ZK_SERVICE = PropertyConfigHelper.getZkService();
    private static int ZK_SESSION_TIME_OUT = PropertyConfigHelper.getZkSessionTimeout();
    private static int ZK_CONNECTION_TIMEOUT = PropertyConfigHelper.getZkConnectionTimeout();

    private static String ROOT_PATH = "/config_register";
    private static String PROVIDER_TYPE = "provider";
    private static String INVOKER_TYPE = "consumer";

    private static volatile ZkClient zkClient = null;

    private RegisterCenter() { }

    public static RegisterCenter singleton() {
        return registerCenter;
    }

    /**
     * zk连接Client
     *
     * @return
     */
    private ZkClient connectZk() {
        if(zkClient == null) {
            zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIMEOUT, new SerializableSerializer());
        }
        return zkClient;
    }

    /**-----------------------------for provider------------------*/
    @Override
    public void registerProvider(List<ProviderService> serviceMetaData) {
        if(CollectionUtils.isEmpty(serviceMetaData)) {
            return;
        }
        // 连接ZK,注册服务
        synchronized (RegisterCenter.class) {
            for(ProviderService provider : serviceMetaData) {
                String serviceItfKey = provider.getServiceItf().getName();
                List<ProviderService> providers = providerServiceMap.get(serviceItfKey);
                if(providers == null) {
                    providers = Lists.newArrayList();
                }
                providers.add(provider);
                //把注册的服务加入
                providerServiceMap.put(serviceItfKey, providers);
            }

            zkClient = connectZk();

            // 创建ZK命名空间/当前部署应用APP命名空间
            String APP_KEY = serviceMetaData.get(0).getAppKey();
            String ZK_PATH = ROOT_PATH + "/" + APP_KEY;
            boolean exist = zkClient.exists(ZK_PATH);
            if(!exist) {
                zkClient.createPersistent(ZK_PATH, true);
            }

            for(Map.Entry<String, List<ProviderService>> entry : providerServiceMap.entrySet()) {
                // 服务分组
                String groupName = entry.getValue().get(0).getGroupName();
                // 创建服务提供者
                String serviceNode = entry.getKey();
                String servicePath = ZK_PATH + "/" + groupName + "/" + serviceNode + "/" + PROVIDER_TYPE;
                exist = zkClient.exists(servicePath);
                if(!exist) {
                    zkClient.createPersistent(servicePath, true);
                }

                // 创建当前服务器节点
                // 服务端口
                int servicePort = entry.getValue().get(0).getServerPort();
                // 服务权重
                int weight = entry.getValue().get(0).getWeight();
                // 服务工作线程数
                int workerThreads = entry.getValue().get(0).getWorkerThread();
                String localIp = IPHelper.localIp();
                String currentServiceIpNode = servicePath + "/" + localIp + "|" + servicePort
                        + "|" + weight + "|" + workerThreads + "|" + groupName;

                exist = zkClient.exists(currentServiceIpNode);
                if(!exist) {
                    // 创建临时节点
                    zkClient.createEphemeral(currentServiceIpNode);
                }

                // 监听注册服务的变化，同时更新数据到本地缓存
                zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
                    @Override
                    public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                        if (currentChilds == null) {
                            currentChilds = Lists.newArrayList();
                        }

                        List<String> activityServiceIpList = Lists.newArrayList(Lists.transform(currentChilds, new Function<String, String>() {
                            @Override
                            public String apply(String input) {
                                return StringUtils.split(input, "|")[0];
                            }
                        }));
                        // 刷新存活的服务
                        refreshActivityService(activityServiceIpList, providerServiceMap);
                    }
                });

            }
        }
    }

    @Override
    public Map<String, List<ProviderService>> getProviderServiceMap() {
        return providerServiceMap;
    }

    /**-----------------------------for invoker------------------*/
    @Override
    public void subscribe(String serviceName, String remoteAppKey, String groupName) {
        if (MapUtils.isEmpty(serviceMetaData4ConsumerMap)) {
           fetchOrUpdateServiceMetaData(serviceName, remoteAppKey, groupName);
        }
    }

    private void fetchOrUpdateServiceMetaData(String invokerServiceName, String remoteAppKey, String groupName) {
        synchronized (RegisterCenter.class) {
            zkClient = connectZk();
        }

        // 从zk获取服务提供者所有服务列表（这里获取服务所有的列表（是否值得改成应用调用接口的服务？毕竟服务不会很多））
        String providePath = ROOT_PATH + "/" + remoteAppKey + "/" + groupName;
        List<String> providerServices = zkClient.getChildren(providePath);
        logger.info("providerServices:" + JSON.toJSONString(providerServices));
        for (String serviceName : providerServices) {
            // 不是订阅的服务不要
            if(!StringUtils.equals(serviceName, invokerServiceName)) {
                continue;
            }
            String servicePath = providePath + "/" + serviceName + "/" + PROVIDER_TYPE;

            List<ProviderService> providerServiceList = serviceMetaData4ConsumerMap.get(serviceName);
            if(providerServiceList == null) {
                providerServiceList = Lists.newArrayList();
            }

            List<String> ipPathList = zkClient.getChildren(servicePath);
            for(String ipPath : ipPathList) {
                String[] ipNodeInfos = StringUtils.split(ipPath, "|");
                String serverIp = ipNodeInfos[0];
                int serverPort = Integer.parseInt(ipNodeInfos[1]);
                int weight = Integer.parseInt(ipNodeInfos[2]);
                int workerThreads = Integer.parseInt(ipNodeInfos[3]);
                String group = ipNodeInfos[4];

                // 创建服务提供者信息
                ProviderService providerService = new ProviderService();
                try {
                    providerService.setServiceItf(ClassUtils.getClass(serviceName));
                } catch (ClassNotFoundException e) {
                    logger.error("ClassUtils.getClass(" + serviceName +") is error.", e);
                    throw new RuntimeException(e);
                }

                providerService.setServerIp(serverIp);
                providerService.setServerPort(serverPort);
                providerService.setAppKey(remoteAppKey);
                providerService.setWeight(weight);
                providerService.setWorkerThread(workerThreads);
                providerService.setGroupName(group);
                providerServiceList.add(providerService);
            }
            serviceMetaData4ConsumerMap.put(serviceName, providerServiceList);

            // 监听子节点变化，刷新存活服务，这里只考虑了服务减少的情况，对于增加的情况没有考虑，后面补吧
            zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
                @Override
                public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                    if(currentChilds == null) {
                        currentChilds = Lists.newArrayList();
                    }
                    currentChilds = Lists.newArrayList(Lists.transform(currentChilds, new Function<String, String>() {
                        @Override
                        public String apply(String input) {
                            return StringUtils.split(input, "|")[0];
//                            return input;
                        }
                    }));
                    refreshActivityService(currentChilds, serviceMetaData4ConsumerMap);
                }
            });
        }
    }

    @Override
    public Map<String, List<ProviderService>> getServiceMetaDataMap4Consume() {
        return serviceMetaData4ConsumerMap;
    }

    @Override
    public void registerInvoker(InvokeService invokeService) {
        if (invokeService == null) {
            return;
        }

        synchronized (RegisterCenter.class) {
            zkClient = connectZk();

            // 创建 ZK命名空间/当前部署应用命名空间
            boolean exist = zkClient.exists(ROOT_PATH);
            if(!exist) {
                zkClient.createPersistent(ROOT_PATH, true);
            }

            String remoteAppKey = invokeService.getRemoteAppKey();
            String groupName = invokeService.getGroupName();
            String serviceNode = invokeService.getServiceItf().getName();
            String appKey = invokeService.getAppKey();
            String servicePath = ROOT_PATH + "/" + remoteAppKey + "/" + groupName + "/"
                    + serviceNode + "/" + appKey + "_" + INVOKER_TYPE;
            exist = zkClient.exists(servicePath);
            if(!exist) {
                zkClient.createPersistent(servicePath, true);
            }

            String localIp = IPHelper.localIp();
            String currentServiceIpNode = servicePath + "/" + localIp;
            exist = zkClient.exists(currentServiceIpNode);
            if (!exist) {
                // 创建临时节点
                zkClient.createEphemeral(currentServiceIpNode);
            }
        }
    }

    /**
     * 利用ZK自动刷新当前存活的服务提供者列表数据
     *
     * @notice 这里只考虑了服务减少的情况，没有考虑扩容的情况，扩容只能通过重启服务
     * 或
     * 起一个定时任务重新捞一把
     *
     * @param serviceIpList
     * @param serviceMetaDataMap
     */
    private void refreshActivityService(List<String> serviceIpList, Map<String, List<ProviderService>> serviceMetaDataMap) {
        if(serviceIpList == null) {
            serviceIpList = Lists.newArrayList();
        }

        Map<String, List<ProviderService>> currentServiceMetaDataMap = Maps.newConcurrentMap();
        for(Map.Entry<String, List<ProviderService>> entry : serviceMetaDataMap.entrySet()) {
            String key = entry.getKey();
            List<ProviderService> providerServices = entry.getValue();

            List<ProviderService> currentServiceMetaDataModeList = currentServiceMetaDataMap.get(key);
            if(currentServiceMetaDataModeList == null) {
                currentServiceMetaDataModeList = Lists.newArrayList();
            }

            for(ProviderService serviceMetaData : providerServices) {
                // 如果当前缓存节点包含的变更的子列表内，则加入
                // 这里应该还有可能会扩容
                if(serviceIpList.contains(serviceMetaData.getServerIp())) {
                    currentServiceMetaDataModeList.add(serviceMetaData);
                }
            }
            currentServiceMetaDataMap.put(key, currentServiceMetaDataModeList);
        }
        // 清空当前缓存
        serviceMetaDataMap.clear();
        logger.info("currentServiceMetaDataMap：" + JSON.toJSONString(currentServiceMetaDataMap));
        serviceMetaDataMap.putAll(currentServiceMetaDataMap);
    }

    /** -----------------------------for Governance ---------------------------------*/
    @Override
    public Pair<List<ProviderService>, List<InvokeService>> queryProvidersAndInvokers(String serviceName, String appKey) {
        // 服务消费者列表
        List<InvokeService> invokeServices = Lists.newArrayList();
        // 服务提供者列表
        List<ProviderService> providerServices = Lists.newArrayList();

        synchronized (RegisterCenter.class) {
            zkClient = connectZk();
        }

        String parentPath = ROOT_PATH + "/" + appKey;
        // 获取ROOT_PATH + APPKEY注册中心子目录列表
        List<String> groupServiceList = zkClient.getChildren(parentPath);
        if (CollectionUtils.isEmpty(groupServiceList)) {
            return Pair.of(providerServices, invokeServices);
        }

        for (String group : groupServiceList) {
            String groupPath = parentPath + "/" + group;
            // 获取ROOT_PATH + APPKEY + group 注册中心子目录列表
            List<String> serviceList = zkClient.getChildren(groupPath);
            if (CollectionUtils.isEmpty(serviceList)) {
                continue;
            }
            for (String service : serviceList) {
                // 获取ROOT_PATH + APPKEY + group + service 注册中心子目录列表
                String servicePath = groupPath + "/" + service;
                List<String> serviceTypes = zkClient.getChildren(servicePath);
                if (CollectionUtils.isEmpty(serviceTypes)) {
                    continue;
                }

                for (String serviceType : serviceTypes) {
                    // 获取ROOT_PATH + APPKEY + group + service + PROVIDER_TYPE or INVOKER_TYPE注册中心子目录列表
                    String serviceTypePath = servicePath + "/" + serviceType;
                    List<String> serviceTypePaths = zkClient.getChildren(serviceTypePath);
                    if(CollectionUtils.isEmpty(serviceTypePaths)) {
                        continue;
                    }

                    if(StringUtils.equals(serviceType, PROVIDER_TYPE)) {
                        for (String provider : serviceTypePaths) {
                            String[] providerNodeArr = StringUtils.split(provider, "|");

                            ProviderService providerService = new ProviderService();
                            providerService.setAppKey(appKey);
                            providerService.setGroupName(group);
                            providerService.setServerIp(providerNodeArr[0]);
                            providerService.setServerPort(Integer.parseInt(providerNodeArr[1]));
                            providerService.setWeight(Integer.parseInt(providerNodeArr[2]));
                            providerService.setWorkerThread(Integer.parseInt(providerNodeArr[3]));
                            providerServices.add(providerService);
                        }
                    } else if (StringUtils.endsWith(serviceType , INVOKER_TYPE)) {
                        String invokerAppKey = StringUtils.split("_")[0];
                        for (String invoker : serviceTypePaths) {
                            InvokeService invokeService = new InvokeService();
                            invokeService.setRemoteAppKey(appKey);
                            invokeService.setGroupName(group);
                            invokeService.setInvokerIp(invoker);
                            invokeService.setAppKey(invokerAppKey);
                            invokeServices.add(invokeService);
                        }
                    }
                }
            }
        }

        return Pair.of(providerServices, invokeServices);
    }
}
